package cn.lagou.part01

import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

case class Ip(startIp: String, endIp: String, address: String)

object FindArea {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
      .setAppName(this.getClass.getCanonicalName)
      .setMaster("local[4]")

    val sc: SparkContext = new SparkContext(conf)
    sc.setLogLevel("warn")

    // http log信息
    val ipLogsRDD = sc.textFile("data/http.log").map(_.split("\\|")(1))

    // ip信息
    val ipRDD: RDD[Ip] = sc.textFile("data/ip.dat").map(line => {
      val arr: Array[String] = line.split("\\|")
      var area = if (arr(7).isEmpty) arr(6) else arr(7)
      Ip(arr(0), arr(1), area)
    })
    //    ipRDD.collect().foreach(println)

    // 广播变量
    val brIpInfo: Broadcast[Array[(Long, Long, String)]] = sc.broadcast(ipRDD.map(x => (ip2Long(x.startIp), ip2Long(x.endIp), x.address)).collect())
    //    brIpInfo.value.foreach(println)
    ipLogsRDD
      .map(x => {
        val index = binarySearch(brIpInfo.value, ip2Long(x))
        if (index != -1) brIpInfo.value(index)._3
        else "nil"
      }).map(x => (x, 1))
      .reduceByKey(_ + _)
      .map(x => s"城市：${x._1}, 访问量：${x._2}")
      .saveAsTextFile("data/part01/result")

    sc.stop()
  }

  def ip2Long(ip: String): Long = {
    val fragments = ip.split("[.]")
    var ipNum = 0L
    for (i <- 0 until fragments.length) {
      ipNum = fragments(i).toLong | ipNum << 8L
    }
    ipNum
  }

  // 二分法匹配ip规则
  def binarySearch(lines: Array[(Long, Long, String)], ip: Long): Int = {
    var low = 0
    var high = lines.length - 1
    while (low <= high) {
      val middle = (low + high) / 2
      if ((ip >= lines(middle)._1) && (ip <= lines(middle)._2))
        return middle
      if (ip < lines(middle)._1)
        high = middle - 1
      else {
        low = middle + 1
      }
    }
    -1
  }

  def test(): Unit = {
    val strings: Array[String] = "abc|hello".split("\\|")
    strings.foreach(println)
  }

}
